package com.sui.bigdata.rtcadmin.testEnv;

import com.feidee.fdcommon.model.StringResponse;
import com.feidee.fdcommon.util.HttpUtil;
import org.apache.http.Consts;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;


/**
 * @author YongChen
 * @date 2019/11/25 10:08
 * @description
 * @email yong_chen@sui.com
 */
public class submit {
    public static void main(String[] args) throws Exception {


        String url = "http://10.201.7.132:18001/rtc-admin/job?action=submit";
//        String url = "http://10.201.7.132:18001/rtc-admin/job?action=restart";

        String sql = " \n" +
                " CREATE TABLE dwb_ssj_channel_active(\n" +
                "  udid      varchar,\n" +
                "  channel   varchar,\n" +
                "  cid       varchar\n" +
                " )WITH(\n" +
                "    type ='kafka',\n" +
                "    bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                "    offsetReset ='latest',\n" +
                "    groupId = 'dwb_ssj_channel_active_2020/06/15',\n" +
                "    topic ='rthbase.dwb_ssj_channel_active',\n" +
                "    parallelism ='1'\n" +
                " );\n" +
                " \n" +
                "  CREATE TABLE adl_bd_ls_ssj_transaction_side_ADS798 (\n" +
                "    info:qdlx_6 varchar as qdlx_6 ,\n" +
                "    PRIMARY KEY (rowkey), \n" +
                "    PERIOD FOR SYSTEM_TIME \n" +
                ") WITH (\n" +
                "   type = 'hbase',\n" +
                "   zookeeperQuorum = '10.201.7.114:2181,10.201.7.115:2181,10.201.7.113:2181',\n" +
                "   zookeeperParent = '/hbase',\n" +
                "   tableName = 'adl_bd_ls_ssj_transaction_side_ADS798',\n" +
                "   cache ='ALL',\n" +
                "   cacheTTLMs = '3600000'); "  +
                "\n" +
                "CREATE TABLE ssj_transaction(\n" +
                "  suid                  varchar,\n" +
                "  create_time           varchar,\n" +
                "  parent_category_name  varchar,\n" +
                "  fmemo                 varchar\n" +
                " )WITH(\n" +
                "    type ='kafka',\n" +
                "    bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                "    offsetReset ='latest',\n" +
                "    groupId = 'ssj_transaction_2020/06/15',\n" +
                "    topic ='etl_dws.dws_bill_ssj_transaction',\n" +
                "    parallelism ='1'\n" +
                " );\n" +
                " \n" +
                "CREATE TABLE tag1(\n" +
                "     hbase_table varchar,  \n" +
                "     udidmd5 varchar,\n" +
                "     udid varchar,\n" +
                "     devssj_acitve_channel_type varchar\n" +
                "  )WITH(\n" +
                "     type ='kafka',\n" +
                "     bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                "     topic ='rttag.realtime.tag_total',\n" +
                "     parallelism ='1'\n" +
                "  );  \n" +
                "  \n" +
                "  CREATE TABLE tag2(\n" +
                "     hbase_table varchar,  \n" +
                "     udidmd5 varchar,\n" +
                "     sid varchar,\n" +
                "     userssj_lastest_medical_mark_time varchar\n" +
                "  )WITH(\n" +
                "     type ='kafka',\n" +
                "     bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                "     topic ='rttag.realtime.tag_total',\n" +
                "     parallelism ='1'\n" +
                "  ); \n" +
                "  \n" +
                "  CREATE TABLE tag3(\n" +
                "     hbase_table varchar,  \n" +
                "     udidmd5 varchar,\n" +
                "     sid varchar,\n" +
                "     userssj_lastest_travel_mark_time varchar\n" +
                "  )WITH(\n" +
                "     type ='kafka',\n" +
                "     bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                "     topic ='rttag.realtime.tag_total',\n" +
                "     parallelism ='1'\n" +
                "  ); \n" +
                "  \n" +
                "  CREATE TABLE tag4(\n" +
                "     hbase_table varchar,  \n" +
                "     udidmd5 varchar,\n" +
                "     sid varchar,\n" +
                "     userssj_lastest_baby_mark_time varchar\n" +
                "  )WITH(\n" +
                "     type ='kafka',\n" +
                "     bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                "     topic ='rttag.realtime.tag_total',\n" +
                "     parallelism ='1'\n" +
                "  ); \n" +
                " insert into tag1\n" +
                " select \n" +
                "     'ssj_dev_tagging' as hbase_table,\n" +
                "     md5(t1.udid) as udidmd5, \n" +
                "\t t1.udid,\n" +
                "\t case when t2.qdlx_6 ='ASO渠道'    then 'ASO渠道'\n" +
                "\t      when t2.qdlx_6 ='外部合作'   then '外部合作渠道'\n" +
                "\t\t  when t2.qdlx_6 ='运营系统部' then 'QQ运动'\n" +
                "\t\t  else '其他正常渠道' end as devssj_acitve_channel_type\n" +
                " from (select udid ,md5(concat(channel,cid)) as rowkey \n" +
                "       from  dwb_ssj_channel_active ) t1\n" +
                " left join adl_bd_ls_ssj_transaction_side_ADS798 t2\n" +
                " on t1.rowkey = t2.rowkey\n" +
                " where NotBlank(t1.udid);\n" +
                "\n" +
                " insert into tag2\n" +
                " select \n" +
                "     'ssj_user_tagging' as hbase_table,\n" +
                "     md5(suid) as rowkey,\n" +
                "     suid  as suid,\n" +
                "     max(create_time) as userssj_lastest_medical_mark_time\n" +
                " from ssj_transaction\n" +
                " where \t  parent_category_name in ('医疗保健')\n" +
                "       or fmemo like '%挂号%'  \n" +
                "       or fmemo like '%药品%'   \n" +
                "       or fmemo like '%医院%'  \n" +
                "       or fmemo like '%打针%'\n" +
                " group by suid;\n" +
                " \t  \n" +
                " insert into tag3\n" +
                " select \n" +
                "     'ssj_user_tagging' as hbase_table,\n" +
                "     md5(suid) as rowkey,\n" +
                "     suid  as suid,\n" +
                "     max(create_time) as userssj_lastest_travel_mark_time\n" +
                " from ssj_transaction\n" +
                " where    parent_category_name in ('行车交通')\n" +
                "       or fmemo like '%机票%'  \n" +
                " \t   or fmemo like '%高铁%'  \n" +
                " \t   or fmemo like '%车票%'  \n" +
                " \t   or fmemo like '%旅游%'  \n" +
                " \t   or fmemo like '%出行%'  \n" +
                " \t   or fmemo like '%酒店%'  \n" +
                " \t   or fmemo like '%住宿%'  \n" +
                " group by suid;\t\n" +
                " \n" +
                " insert into tag4\n" +
                " select \n" +
                "     'ssj_user_tagging' as hbase_table,\n" +
                "     md5(suid) as rowkey,\n" +
                "     suid  as suid,\n" +
                "     max(create_time) as userssj_lastest_baby_mark_time\n" +
                " from ssj_transaction\n" +
                " where    fmemo like '%宝宝%'  \n" +
                " \t   or fmemo like '%婴儿%'  \n" +
                " \t   or fmemo like '%纸尿片%'\n" +
                " \t   or fmemo like '%早教%'\n" +
                " \t   or fmemo like '%奶嘴%' \n" +
                " group by suid;  \n" +
                "\t  \n" ;


        String data = "{\"execSql\":\"" + sql +
                "\",\"jobName\":\"ssj_transaction_ADS-798\"," +
                "\"developer\":\"yong_chen\"," +
                "\"alarmGroupPhone\":\"\"," +
                "\"parallelism\":\"1\"," +
                "\"queue\":\"root.data_team\"," +
                "\"jobManagerMemory\":\"800\"," +
                "\"taskManagerMemory\":\"1024\"," +
                "\"extendParam\":{\"sql.checkpoint.interval\":30000,\"sql.select.sink\":\"console\",\"state.backend\":\"ROCKSDB\",\"state.checkpoints.dir\":\"hdfs://nameservice1/user/data_team/flink/checkpoints/vis_rt_devssj_3tag_20200423\"},"+
                "\"slots\":\"1\"," +
                "\"description\":\"这是一个flink程序\"," +
                "\"savepoint\":\"true\"}";

        System.out.println(data);
        StringEntity stringEntity = new StringEntity(data, Consts.UTF_8);
        stringEntity.setContentEncoding("UTF-8");
        stringEntity.setContentType("application/json");

        HttpPost httpPost=new HttpPost(url);

        RequestConfig requestConfig = RequestConfig.custom().setConnectTimeout(500000).build();
        httpPost.setEntity(stringEntity);
        httpPost.setConfig(requestConfig);
        StringResponse response = HttpUtil.post(httpPost);
        System.out.println(">>>>> "+response.getResponseBody());
    }
}

